/* file name  : STUNTEndpoint.java
 * authors    : Saikat Guha
 * created    : Sat 24 Sep 2005 02:03:45 PM EDT
 * copyright  : GNU Lesser General Public License version 2.1 or later
 *
 * Copyright (C) 2005 Saikat Guha <saikat@cs.cornell.edu>
 *
 * This library is free software; you can redistribute it and/or modify it under the terms
 * of the GNU Lesser General Public License as published by the Free Software Foundation;
 * either version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
 *  without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License along with this
 * library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330,
 * Boston, MA 02111-1307 USA
 *
 */
package net.nutss.stunt;

import java.net.*;
import java.util.*;
import java.io.*;
import java.util.concurrent.*;
import net.nutss.*;
import net.nutss.signaling.*;

import static net.nutss.Control.log;
import static net.nutss.stunt.STUNTMessage.Stage.*;

/**
 * Implements a STUNT TCP endpoint. The same class is used for both TCP server sockets and
 * TCP client sockets. accept() and connect() operations are asyncronous and reported via
 * a callback mechanism. Instead of IP addresses and ports, the user-visible address is a
 * URI which can be unique even in the presence of NATs.
 *
 * @author Saikat Guha
 * @version %I%, %G%
 */
public class STUNTEndpoint  {

    private enum Type {PROTO_TCP, PROTO_UDP};

    /**
     * Real worker thread
     */
    private Worker worker;

    /**
     * The real class that implements the STUNT endpoint. Encapsulated to hide public
     * methods of this class from access by user functions.
     *
     * @author Saikat Guha
     * @version %I%, %G%
     */
    public void setTerm(boolean tr)
    {
    	if(worker!=null)worker.setTerm(tr);
    	
    }
    public boolean getTerm()
    {
    	return (worker!=null)?worker.getTerm():true;
    	
    }
    private static class Worker extends Dispatch {

        /**
         * Used for synchronizing between the thread creating the endpoint
         * and the thread performing the equivalent of the bind operation.
         */
        private Semaphore semaphore;

        /**
         * local endpoint's URI
         */
        private final URI id;

        /**
         * STUNT servers learned from the signaling infrastructure
         */
        private SocketAddress[] servers;

        /**
         * Tracks the various running continuation/event handlers and the callback
         * object associated with them.
         */
        private HashMap<Continuation, STUNTEventListener> passive, active, keepalive;

        /**
         * Stores the reason why the whole STUNT endpoint was closed. This is different
         * from intermediate exceptions generated as and when connect() operations fail.
         */
        private IOException reason;

        /**
         * Whether the endpoint is closed or not
         */
        private boolean closed;

        /**
         * Creates a new worker bound to the given signaling context and local address
         *
         * @param context signaling channel
         * @param id local ID
         */
        private Worker(SignalingContext context, URI id) {
            super(context);
            this.id = id;
            semaphore = new Semaphore(0);
            passive = new HashMap<Continuation, STUNTEventListener>();
            active = new HashMap<Continuation, STUNTEventListener>();
            keepalive = new HashMap<Continuation, STUNTEventListener>();
        }

        public void callback(Continuation k, Object o) {
            STUNTEventListener cb;
            if ((cb = passive.get(k)) != null) {
                // an incoming connect() attempt either succeeded or failed.
                // If it succeeded, notify the callback
                // If it failed, the callback doesn't care either way
                // In any case, the continuation that was performing the accept
                // is dead.
                if (o instanceof STUNTSocket) cb.acceptTCPHandler(((STUNTSocket)o).s, ((STUNTCont)k).dst);
                /*
                 by casper not supported
                else if (o instanceof DatagramChannel) cb.acceptUDPHandler((DatagramChannel)o, ((STUNCont)k).dst);
                 */
                passive.remove(k);
            } else if ((cb = active.get(k)) != null) {
                // an outgoing connect() succeeded or failed. Notify the callback
                // in both cases. Reap the deal continuation.
                if (o instanceof STUNTSocket) cb.connectTCPHandler(((STUNTSocket)o).s);
                else if (o instanceof STUNTUDPSocket) cb.connectUDPHandler((STUNTUDPSocket)o);
                else if (o instanceof Exception) cb.errorHandler((Exception) o);
                active.remove(k);
            } else if ((cb = keepalive.get(k)) != null) {
                // keepalives are signaling context specific. Can't assume much about
                // them. If they give us some STUNT server addresses, use them. If there
                // is an error, reap them. Also, if we got server addresses, we bind() to
                // the given URI succeeded -- inform the creater thread if it is still waiting.
                if (o instanceof SocketAddress[]) {
                    servers = (SocketAddress[]) o;
                    semaphore.release();
                } else if (o instanceof Exception) {
                    cb.errorHandler((Exception) o);
                    keepalive.remove(k);
                }
            } else {
                return;
            }
        }

        /**
         * Checks whether the endpoint has already by closed. This should be called
         * with the synchronization lock on this object held.
         *
         * @throws IOException if the endpoint has been closed.
         */
        private void checkClosed() throws IOException {
            if (closed) {
                IOException e = new IOException("Endpoint closed.");
                e.initCause(reason);
                log.throwing("Worker", "checkClosed", e);
                throw e;
            }
        }

        /**
         * Stores the master callback for endpoints in listen mode. Each time
         * a new connection attemp arrives, a mini continuation is created and
         * associated with a copy of this callback object.
         */
        private STUNTEventListener listenCB;

        /**
         * Marks this endpoint as willing to receive incoming connections
         *
         * @param cb the callback functions to call for accept
         * @throws IOException if the socket has already been closed, or already listening
         */
        void listen(STUNTEventListener cb) throws IOException {
            synchronized(this) {
                checkClosed();
                if (listenCB != null) throw new IOException("Already listening.");
                listenCB = cb;
            }
        }

        /**
         * Sets the reason why the endpoint was closed.
         *
         * @param e the reason
         */
        private void reason(Exception e) {
            if (reason == null) {
                reason = new IOException("Endpoint closed.");
                reason.initCause(e);
            }
        }

        public void handleMessage(SignalingMessage msg) {
            synchronized(this) {
                // Make sure the proxy message is an INVITE and not a stray
                // error response for an old invite or something.
                if (msg.hasPayload() && msg.getPayload() instanceof STUNTMessage &&
                        !msg.isError() && listenCB != null) {
                    if (((STUNTMessage)msg.getPayload()).stage == STUNT_INVITE) {
                        // If user/app confimation is required before attempting to establish
                        // a connection, this would be a good place to check. The listenCB
                        // that was passed from the app earlier can have an 'boolean allowURI(URI remote)'
                        // function.
                        Continuation k = new STUNTCont(msg.getSource(), msg.getDestination(), sig, servers);
                        passive.put(k, listenCB);
                        addContinuation(k, msg);
                    }
                    /*
                    by casper not supported
                    if (((STUNTMessage)msg.getPayload()).stage == STUN_INVITE) {
                        try {
                            // If user/app confimation is required before attempting to establish
                            // a connection, this would be a good place to check. The listenCB
                            // that was passed from the app earlier can have an 'boolean allowURI(URI remote)'
                            // function.
                            Continuation k = new STUNCont(msg.getSource(), msg.getDestination(), sig, servers, null);
                            passive.put(k, listenCB);
                            addContinuation(k, msg);
                        } catch (IOException e) {
                        }
                    }
                    */
                }
            }
        }

        /**
         * Initiate an outgoing connection
         *
         * @param dst destination ID
         * @param cb callback for success or failure
         * @throws IOException if the endpoint has been closed
         */
        void connect(URI dst, STUNTEventListener cb, Type t) throws IOException {
            synchronized(this) {
                checkClosed();
                Continuation k = null;
                switch (t) {
                    case PROTO_TCP:
                        k = new STUNTCont(dst, id, sig, servers);
                        break;

                        /*
                        by casper not supported
                    case PROTO_UDP:
                        k = new STUNCont(dst, id, sig, servers, null);
                        break;
                        */

                    default:
                        throw new IllegalStateException("Protocol not supported");
                }
                active.put(k, cb);
                addContinuation(k, null);
            }
        }

        public void run() {
            try {
                // First start a KeepAlive continuation as a result of
                // registering/binding the ID of the local endpoint with
                // the proxy infrastructure
                Continuation k = sig.register(id,sel);

                if (k != null) {
                    keepalive.put(k, new STUNTEventAdapter() {

                        /**
                         * Handles errors on the keepalive continuation. If the
                         * continuation terminated, we will soon lose our binding.
                         * Terminate the endpoint as a result.
                         *
                         * @param e the error.
                         */
                        public void errorHandler(Exception e) {
                            reason(e);
                            interrupt();
                        }
                    });
                    addContinuation(k, null);
                }

                // Start processing events. If this returns, endpoint is dead.
                super.run();

                // Should never reach here under normal operations. This
                // signifies that all continuations (even keepalive!) have
                // exited without reporting an error.
                log.fine("No more continuations");
                reason(new Exception("All continuations exited"));

            } catch (Exception e) {
                log.throwing("Worker", "run", e);
                reason(e);
            }

            synchronized(this) {
                // be sure to report endpoint death to all callbacks
                // avoid race conditions.
                closed = true;
                reason(new InterruptedException("Interrupted."));
                for(STUNTEventListener l : passive.values()) { l.errorHandler(reason); }
                for(STUNTEventListener l : active.values()) { l.errorHandler(reason); }
            }

            // the thread that created this worker may still be waiting to get a
            // success or failure report; which will never come since the endpoint is
            // dead. Kick that thread back into action.
            semaphore.release();
        }

    }

    /**
     * Creates a new STUNT endpoint bound to the local user and hostname.
     *
     * @throws IOException if endpoint couldn't bind
     */
    public STUNTEndpoint() throws IOException {
        this(null);
    }

    /**
     * Creates a new STUNT endpoint bound to the given URI. If id is null, binds
     * to a random name based on the local username and hostname.
     *
     * @param id the local address
     * @throws IOException if endpoint couldn't bind
     */
    public STUNTEndpoint(URI id) throws IOException {
        try {
            if (id == null)
                id = new URI(System.getenv("USER")+((int)(Math.random()*10000000))+"@84.234.17.41");
            worker = new Worker(new JSlingProvider(), id);
            worker.setDaemon(true);
            worker.start();

            // Wait for the worker to signal us whether or not the
            // proxy communication worked out.
            log.finer("Acquiring semaphore");
            worker.semaphore.acquire();
            log.finer("Acquired semaphore");
            if (worker.reason != null) throw worker.reason;
        } catch (Exception e) {
            if (worker != null) worker.interrupt();
        }
    }

    /**
    * Marks this endpoint as willing to receive incoming connections
    *
    * @param cb the callback functions to call for accept()
    * @throws IOException if the socket has already been closed, or already listening
     */
    public void listen(STUNTEventListener cb) throws IOException {
        worker.listen(cb);
    }

    /**
    * Initiate an outgoing TCP connection
    *
    * @param dst destination ID
    * @param cb callback for success or failure
    * @throws IOException if the endpoint has been closed
     */
    public void connectTCP(URI dst, STUNTEventListener cb) throws IOException {
        worker.connect(dst, cb, Type.PROTO_TCP);
    }

    /**
    * Initiate an outgoing UDP connection
    *
    * @param dst destination ID
    * @param cb callback for success or failure
    * @throws IOException if the endpoint has been closed
     */
    public void connectUDP(URI dst, STUNTEventListener cb) throws IOException {
        worker.connect(dst, cb, Type.PROTO_UDP);
    }

    /**
     * Closes the endpoint.
     */
    public void close() {
        worker.interrupt();
    }

}
